package cn.gwm.flink.streaming.sink.hbase;

import cn.gwm.flink.streaming.constant.BaseFields;
import cn.gwm.utils.StringUtil;
import cn.hutool.json.JSONObject;
import lombok.AllArgsConstructor;
import org.apache.flink.streaming.api.datastream.DataStream;

/**
 * @Author: liangguang
 * @Date: 2022-08-25
 * @Description: hbase 落地
 */
@AllArgsConstructor
public class HbaseSinkClient {

    /**
     * hbase 落地  默认 rowkwy为 deviceId + item_time 反转
     *
     * @param hbaseStream
     * @param tableName
     * @param columnFamily
     */
    public static void hbaseSink(DataStream<JSONObject> hbaseStream, String tableName, String columnFamily) {
        hbaseSink(hbaseStream, tableName, columnFamily, BaseFields.vin, BaseFields.tid);
    }

    /**
     * hbase 落地
     *
     * @param hbaseStream
     * @param tableName
     * @param columnFamily
     * @param rowkeyVin    rowkey 的 vin 字段
     * @param rowkeyTid    rowkey 的 tid 字段 用来反转
     */
    public static void hbaseSink(DataStream<JSONObject> hbaseStream, String tableName, String columnFamily, String rowkeyVin, String rowkeyTid) {
        hbaseStream.map(item -> {
            String rowkey = item.getStr(rowkeyVin) + StringUtil.reverse(item.getStr(rowkeyTid));
            HbaseBean bean = new HbaseBean()
                    .setRowKey(rowkey)
                    .setTableName(tableName)
                    .setColumnFamily(columnFamily)
                    .setDatamap(item);
            return bean;
        }).addSink(new HbaseSinkFunction()).name("hbaseSink");
    }

    /**
     * hbase 落地
     *
     * @param hbaseStream
     * @param tableName
     * @param columnFamily
     *    rowkey 的 vin 字段
     *    rowkey 的 tid 字段 用来反转
     */
    public static void hbaseSink(DataStream<JSONObject> hbaseStream, String tableName, String columnFamily,String name) {
        hbaseStream.map(item -> {
            String rowkey = item.getStr(BaseFields.vin) + StringUtil.reverse(item.getStr(BaseFields.tid));
            HbaseBean bean = new HbaseBean()
                    .setRowKey(rowkey)
                    .setTableName(tableName)
                    .setColumnFamily(columnFamily)
                    .setDatamap(item);
            return bean;
        }).addSink(new HbaseSinkFunction()).name(name);
    }

}
